package com.bigdata.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo4_SubmitDemo {

  def main(args: Array[String]): Unit = {

    // 创建spark配置对象
    val conf: SparkConf = new SparkConf()
      .setAppName("WordCount") // 设置应用名称

    // 需要一个SparkContext对象
    val sc = new SparkContext(conf)
    // 设置日志的打印级别
    sc.setLogLevel("error")

    // 读取数据，集群模式，路径指的是HDFS上的路径
    val linesRDD: RDD[String] = sc.textFile("/data/words.txt", args(0).toInt)

    // 获取分区数, 就相当于并行度，driver里执行的
    println("getNumPartitions----" + linesRDD.getNumPartitions)

    /**
     * spark 中提供了很多算子，比scala的集合函数还要丰富
     * rdd算子的代码是在哪里执行的？？？？
     * 在Executor中的线程池里执行的
     *
     */
    val wordRDD: RDD[String] = linesRDD.flatMap(line => {
      line.split(",")
    })

    val pairRDD: RDD[(String, Int)] = wordRDD.map(word => (word, 1))

    // 按照key分区聚合，指定value的聚合逻辑
    val result: RDD[(String, Int)] = pairRDD.reduceByKey((v1: Int, v2: Int) => v1 + v2)

    println(result.getNumPartitions)
    // action 触发，算子的代码依然在Executor中执行
    result.foreach(println)

    // HDFS上的目录
    result.saveAsTextFile("/data/wc_output")
  }

}
